package com.lagou.hbase.processor;

import com.google.common.base.Objects;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTableWrapper;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.security.access.AccessController;
import org.apache.hadoop.hbase.security.visibility.VisibilityController;
import org.apache.hadoop.hbase.tool.WriteSinkCoprocessor;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;
import org.junit.Before;
import org.mortbay.util.ajax.JSON;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * @Author: Hugh
 * @Date: 2020/9/17
 */

public class HomeWorkProcessor extends BaseRegionObserver {


    private Configuration conf = null;
    private Connection conn = null;
    private Table userRela = null;

    private static final Log LOG = LogFactory.getLog(HomeWorkProcessor.class);
    /**
     * 删除好友操作实现（好友关系双向，一方删除好友，另一方也会被迫删除好友）
     * @param e
     * @param delete
     * @param edit
     * @param durability
     * @throws IOException
     *
     * hbase(main):022:0> delete 'user_relationship', 'uuid1', 'friends:uuid3'
     * 0 row(s) in 0.0440 seconds
     * hbase(main):022:0> delete 'user_relationship', 'uuid3', 'friends:uuid1'
     */
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //获取好友关系表table对象
        if(null == userRela){
            //final HTable t2 = (HTable) e.getEnvironment().getTable(TableName.valueOf("t2"));
            //HTableWrapper cannot be cast to org.apache.hadoop.hbase.client.HTable
            init();
        }
        try {
            // 首先查看是否对于friends列进行了指定删除
            List<Cell> cells = delete.getFamilyCellMap().get(
                    Bytes.toBytes("friends"));
            LOG.error("HomeWorkProcessorCells:{}" + cells.toString());
            if (CollectionUtils.isEmpty(cells)) {
                userRela.close();
                return;
            }
            for (Cell cell : cells) {
                //获取到firends下的第一个cell
//            Cell cell = cells.get(0);
                // 创建 uid2， 并设置需要删除的 column
                String needDeletRowKey = new String(CellUtil.cloneQualifier(cells.get(0)));;
                if(!isPresent(needDeletRowKey, userRela, delete.getRow())){
                    continue;
                }
                LOG.error("HomeWorkProcessorCell:{}" + needDeletRowKey);
                Delete otherUserDelete = new Delete(CellUtil.cloneQualifier(cell));
                /**
                 * 删除代码示例:
                 *  final Delete delete1 = new Delete(Bytes.toBytes("uuid1"));
                 *  delete1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes("uuid3"));
                 */
                otherUserDelete.addColumns(Bytes.toBytes("friends"), delete.getRow());
                //执行
                userRela.delete(otherUserDelete);
            }

        }catch (Exception ex){
            userRela.close();
        }
        userRela.close();
    }


    private void init() throws IOException {
        //获取一个配置文件对象
        conf = HBaseConfiguration.create();
        //由于启动的是mac的zk，需要配置本地的zk
        conf.set("hbase.zookeeper.quorum", "192.168.1.3");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //通过conf获取到hbase集群的连接
        conn = ConnectionFactory.createConnection(conf);
        //需要获取一个table对象
        userRela = conn.getTable(TableName.valueOf("user_relationship"));
    }

    private Boolean isPresent(String rowkey, Table userRela, byte[] rowkey2)throws IOException {
        //准备get对象
        final Get get = new Get(Bytes.toBytes(rowkey));
        //指定查询某个列族或者列
//        get.addFamily(Bytes.toBytes("friends"));
        get.addColumn(Bytes.toBytes("friends"), rowkey2);
        //执行查询
        final Result result = userRela.get(get);
        //获取到result中所有cell对象
        final Cell[] cells = result.rawCells();
        return null != cells && cells.length > 0;
    }
}
